package com.leenmvc.core.mqtt.service;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.leenmvc.core.mqtt.entity.TopicMappingInfo;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

@Component
public class PushCallbackService implements MqttCallback, MqttCallbackExtended {

    private final static Logger log = LoggerFactory.getLogger(PushCallbackService.class);


    public static Map<String, TopicMappingInfo> topicMapping = new HashMap<>();

    private MqttConnectOptions mqttConnectOptions;

    private MqttClient mqttClient;

    private int[] Qos;

    private String[] topics;

    public int[] getQos() {
        return Qos;
    }

    public void setQos(int[] qos) {
        Qos = qos;
    }

    public String[] getTopics() {
        return topics;
    }

    public void setTopics(String[] topics) {
        this.topics = topics;
    }

    public MqttConnectOptions getMqttConnectOptions() {
        return mqttConnectOptions;
    }

    public void setMqttConnectOptions(MqttConnectOptions mqttConnectOptions) {
        this.mqttConnectOptions = mqttConnectOptions;
    }

    public MqttClient getMqttClient() {
        return mqttClient;
    }

    public void setMqttClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }

    public PushCallbackService(MqttClient mqttClient, MqttConnectOptions mqttConnectOptions) {
        this.mqttClient = mqttClient;
        this.mqttConnectOptions = mqttConnectOptions;
    }

    public PushCallbackService(){

    }

    @Override
    public void connectComplete(boolean b, String s) {
        try {
//            LogFactory.getGlobalLog().info("连接成功!正在订阅...");
            log.info("连接成功!正在订阅...");
            mqttClient.subscribe(topics, Qos);
            for (String topic : topics) {
                log.info("订阅{}成功" , topic );
//                LogFactory.getGlobalLog().info("订阅" + topic + "成功");
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void connectionLost(Throwable throwable) {
        while (true) {
            try {
                throwable.printStackTrace();
                //如果没有发生异常说明连接成功，如果发生异常，则死循环
                Thread.sleep(1000);
                //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
                if (!mqttClient.isConnected()) {
                    log.warn("mqtt正在重连....");
                    mqttClient.connect(mqttConnectOptions);
                    log.warn("mqtt重连成功");
                } else {
                    //这里的逻辑是如果连接成功就重新连接
                    mqttClient.disconnect();
                    log.warn("mqtt正在重连....");
                    mqttClient.connect(mqttConnectOptions);
                    log.info("重连成功");
                }
                break;
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        String msg = new String(mqttMessage.getPayload());
        log.info("收到topic：{}",topic);
        log.info("收到消息：{}",msg);
        String[] split = topic.split("/");
        String feat =  split[split.length-1];
        TopicMappingInfo topicMappingInfo = topicMapping.get(topic);
        Method method = topicMappingInfo.getMethod();
        Class<?>[] parameterTypes = method.getParameterTypes();
        try {
            method.invoke(topicMappingInfo.getObj(), JSONObject.parseObject(msg, parameterTypes[0]));
        } catch (JSONException e) {
            log.error(e.getMessage(), e);
        } catch (IllegalAccessException e) {
            log.error(e.getMessage(), e);
        } catch (InvocationTargetException e) {
            log.error(e.getMessage(), e);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }
}
